package com.atguigu.mq.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeUnit;

@Component
public class MyMessageListener {

    public static final String EXCHANGE_DIRECT = "exchange.direct.order";
    public static final String ROUTING_KEY = "order";
    public static final String QUEUE_NAME  = "queue.order";

    public static final String EXCHANGE_NORMAL = "exchange_normal221126";
    public static final String EXCHANGE_DEAD_LETTER = "exchange_dead_letter221126";

    public static final String ROUTING_KEY_NORMAL = "routing_key_normal221126";
    public static final String ROUTING_KEY_DEAD_LETTER = "routing_key_dead_letter221126";

    public static final String QUEUE_NORMAL = "queue_normal221126";
    public static final String QUEUE_DEAD_LETTER = "queue_dead_letter221126";

    private static final String EXCHANGE_DELAY = "exchange.delayed.221126";
    private static final String ROUTING_KEY_DELAY = "routing.key.delayed.221126";
    private static final String QUEUE_DELAY = "queue.delayed.221126";

    @RabbitListener(bindings = @QueueBinding(
        value = @Queue(
            value = QUEUE_NAME,
            durable = "true",
            autoDelete = "false",

            // 给队列设置额外参数
            // name 属性：参数名。x-message-ttl表示设置队列中消息的过期时间
            // value 属性：参数值。x-message-ttl参数的值以毫秒为单位
            // type 属性：参数类型
            arguments = @Argument(name = "x-message-ttl", value = "10000", type = "java.lang.Integer")),
        exchange = @Exchange(value = EXCHANGE_DIRECT, durable = "true", autoDelete = "false"),
        key = {ROUTING_KEY}
    ))
    public void processConfirm(String content, Message message, Channel channel) throws IOException {
        try {
            // 打印消息，代表要执行的业务功能
            System.out.println("content = " + content);

            // 为了测试消费端限流效果，让线程睡一会儿，慢点执行
            TimeUnit.SECONDS.sleep(5);

            // 业务代码执行成功，则给 RabbitMQ 服务器返回 ACK 消息
            // deliveryTag：当前消息的唯一标识
            long deliveryTag = message.getMessageProperties().getDeliveryTag();

            // multiple：是否连同当前消息之前所有未确认的消息一起确认。所谓“之前”就是比当前 deliveryTag 值小的那些消息
            // true：都确认，false：只确认当前消息本身
            boolean multiple = false;

            // 给 RabbitMQ 服务器返回 ACK 信息，代表消息消费成功
            channel.basicAck(deliveryTag, multiple);

        } catch (Throwable throwable) {

            // 获取参数，了解消息是否被消费过
            // true：表示当前消息曾经被消费过
            // false：表示当前消息没有被消费过
            Boolean redelivered = message.getMessageProperties().getRedelivered();

            if (redelivered) {
                boolean requeue = false;

                // 如果已经被消费过了，则返回 reject 信息，表示拒绝这个消息
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), requeue);

            } else {
                // 如果当前消息没有被消费过，那么放回消息队列，再试一次
                // 设置当前消息是否放回消息队列
                // true：放回消息队列，消息队列会重新投递这个消息，也就是重新执行当前方法
                // false：不放回消息队列，消息处理到此为止
                boolean requeue = true;

                // 给 RabbitMQ 服务器返回 NACK 信息，代表消息消费失败
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, requeue);

            }

        }
    }

//    @RabbitListener(
//        bindings = @QueueBinding(
//            value = @Queue(
//                value = QUEUE_NORMAL,
//                durable = "true",
//                autoDelete = "false",
//
//                // 设定参数：让正常队列中变成死信的消息进入死信队列
//                arguments = {
//                    // 指定死信交换机
//                    @Argument(name = "x-dead-letter-exchange", value = EXCHANGE_DEAD_LETTER),
//
//                    // 指定死信交换机绑定死信队列的路由键
//                    @Argument(name = "x-dead-letter-routing-key", value = ROUTING_KEY_DEAD_LETTER),
//
//                    // 指定当前队列中最多保存多少个消息
//                    @Argument(name = "x-max-length", value = "10", type = "java.lang.Integer"),
//
//                    // 给队列中的消息设置过期时间
//                    @Argument(name = "x-message-ttl", value = "10000", type = "java.lang.Integer")
//                }
//            ),
//            exchange = @Exchange(value = EXCHANGE_NORMAL),
//            key = {ROUTING_KEY_NORMAL}
//        )
//    )
//    public void processMessageReject(Message message, Channel channel) throws IOException {
//        // 直接表示拒绝
//        // channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
//
//        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//    }

    @RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = QUEUE_DEAD_LETTER, durable = "true", autoDelete = "false"),
        exchange = @Exchange(value = EXCHANGE_DEAD_LETTER),
        key = {ROUTING_KEY_DEAD_LETTER}
    ))
    public void processDeadMessage(Message message, Channel channel) throws IOException {
        String content = new String(message.getBody());
        System.out.println("死信队列接收到的消息 = " + content);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    @RabbitListener(
        bindings = @QueueBinding(
            value = @Queue(
                value = QUEUE_DELAY,durable = "true",autoDelete = "false"
            ),
            exchange = @Exchange(
                value = EXCHANGE_DELAY,
                type = "x-delayed-message",
                arguments = @Argument(name = "x-delayed-type", value = "direct")
            ),
            key = {ROUTING_KEY_DELAY}
        )
    )
    public void processDelayedMessage(String messageContent, Message message, Channel channel) throws IOException {
        System.out.println("消息内容 = " + messageContent);
        System.out.println("当前时间 = " + new SimpleDateFormat("hh:mm:ss").format(new Date()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}
